package com.bosi.message.impl;

import com.bosi.message.Message;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 异步消息服务
 */
@Slf4j
public class AsyncMessageService extends SimpleMessageService {
    private final AsyncOptions opt;

    protected final ExecutorService executorService;
    protected final LinkedBlockingQueue<Message> queue = new LinkedBlockingQueue<>(100000);

    public AsyncMessageService(){
        this(new AsyncOptions());
    }

    public AsyncMessageService(AsyncOptions opt){
        this.opt = opt;
        executorService = Executors.newFixedThreadPool(opt.nThread,new AsyncThreadFactory());
        for (int i = 0; i < opt.nThread; i++) {
            executorService.submit(() -> head());
        }
    }

    private void head() {
        while (true){
            try {
                Message take = queue.take();
                super.publish(take);
            } catch (InterruptedException e) {
                break;
            } catch (Exception e){
                log.warn("异步发送消息失败",e);
            }
        }
    }

    @Override
    public void publish(Message message) {
        queue.add(message);
    }

    public AsyncOptions getOpt() {
        return opt;
    }

    private static class AsyncThreadFactory implements ThreadFactory{
        AtomicInteger index = new AtomicInteger();

        @Override
        public Thread newThread(Runnable r) {
            Thread thread = new Thread(r);
            thread.setName("AsyncMessageService-thread-" + index.incrementAndGet());
            return thread;
        }
    }
}
